feat: support Spark 4 EmptyRelationExec as a native input - #5821
Conversation
github-actions
Bot
added
enhancement
andygrove
left a comment
andygrove
left a comment
There was a problem hiding this comment.
Thanks for this. I checked the branch out and ran it on Spark 4.1.3, plus a compile only pass on Spark 3.5 with Scala 2.12. Your seven new CometExecSuite tests and the Parquet writer test all pass, and a wider sweep of CometExecSuite, CometParquetWriterSuite, CometAggregateSuite and CometJoinSuite came back with 319 passed and no failures.
I also wrote about forty five throwaway probes of my own, comparing Comet against Spark with the new operator both on and off, over aggregates, every join type, union, limit, sort, window, explode, cross join, broadcast nested loop, complex types and five different write paths. Nothing diverged. The operator itself looks right to me.
It also clearly earns its place on the join elimination shapes. SELECT count(*), sum(l._2) FROM t1 l JOIN (SELECT _1, sum(_2) s FROM t2 WHERE _1 < 0 GROUP BY _1) r ON l._1 = r._1 goes from zero Comet operators today to five with this change, with matching results.
My comments are about test coverage, placement and docs rather than the implementation. The one I would most like your thoughts on is on the AQE test.
| } | ||
| } | ||
|
|
||
| test("EmptyRelationExec discovered by AQE feeds native aggregates and Spark existence joins") { |
There was a problem hiding this comment.
I could not reproduce the scenario from #5819 with Comet at its default configuration. Against a Parquet backed table, SELECT count(*), sum(v) FROM (SELECT _1 % 2 AS k, sum(_2) AS v FROM t WHERE _1 < 0 GROUP BY _1 % 2) produces no EmptyRelationExec at all under Comet. Vanilla Spark produces one, Comet produces neither that nor CometEmptyRelationExec, and the query just runs natively end to end.
The reason looks like AQEPropagateEmptyRelation.getEstimatedRowCount. It only learns a stage's row count from a QueryStageExec or a BaseAggregateExec inside the LogicalQueryStage, and once Comet has converted the aggregate that node is a CometHashAggregateExec, so the rule returns None and no empty relation is ever created.
This test reaches the operator because range() together with spark.comet.sparkToColumnar.enabled=false and spark.comet.shuffle.convertFromSparkPlan.enabled=false keeps the inner aggregate a Spark HashAggregateExec. The shapes that do reach CometEmptyRelationExec at default configs are the join elimination ones. I measured eleven shapes and only inner join and left semi join with an empty right side got there, while sort, limit, window, distinct and sort merge join over an empty stage all produce an EmptyRelationExec under plain Spark and nothing at all under Comet.
Would you consider changing this test to a join shape that reaches the operator without the config overrides? That would cover the path users are actually on, and it would fail if a future change made the operator unreachable again. It also seems worth saying in #5819 that the aggregate case needs the BaseAggregateExec gap addressed separately, since this PR does not close it.
|
|
||
| def emptyRelationClass: Option[Class[_ <: SparkPlan]] = Some(classOf[EmptyRelationExec]) | ||
|
|
||
| def create(logical: LogicalPlan): Option[SparkPlan] = Some(EmptyRelationExec(logical)) |
There was a problem hiding this comment.
create only has one caller, and it is emptyRelation in CometExecSuite. There is a spark/src/test/spark-4.x tree that already holds Spark 4 only suites such as CometWidthBucketSuite and CometShuffle4_0Suite. Would you rather put the new tests there? That would let create come out of both production shims, and it would remove the eight assume(isSpark40Plus) cancellations that Spark 3.4 and 3.5 runs report today. A new suite needs registering in dev/ci/check-suites.py and in both pr_build_linux.yml and pr_build_macos.yml.
If the tests do move, it might also be worth splitting them. Five of the seven build a physical plan by hand and call CometExecRule(spark).apply directly, which is the style used in CometExecRuleSuite and RevertNativeForTransitionHeavyStagesSuite rather than in CometExecSuite.
|
|
||
| # Operator Compatibility | ||
|
|
||
| ## Empty Relations |
There was a problem hiding this comment.
docs/source/user-guide/latest/operators.md describes itself as the complete reference for how Comet handles each Spark physical operator, and EmptyRelationExec is not in any of its tables. Could you add a row there as well, linking to this section and noting the Spark 4.0 and later restriction?
| // AQE can replace the write input with a zero-partition empty relation. Keep | ||
| // Spark's writer, which creates an empty task to preserve the output file schema. | ||
| // The native writer only maps existing partitions and cannot do that yet. | ||
| if (hasEmptyRelationInput(op.child)) { |
There was a problem hiding this comment.
I checked that this guard is load bearing rather than defensive. With it patched out, a fresh path write and a CTAS over an AQE eliminated join both fail with PATH_NOT_FOUND, so it is doing real work on the shapes that actually occur and not only on the aggregate shape in the test.
One thought on how it is written. It keys on finding the operator anywhere in the write subtree, but the hazard is really the write input having zero partitions. hasEmptyRelationInput recurses through QueryStageExec.plan and through exchange children, so it also fires when the empty relation sits under a shuffle and the write input has a perfectly normal partition count. Could this comment name #5303 so that whoever fixes the native writer knows to come back and delete the guard?
| * nor an Arrow reader needs to run. Preserve the zero partitions of EmptyRelationExec so Spark's | ||
| * exchanges continue to control aggregate and join partitioning. | ||
| */ | ||
| case class CometEmptyRelationExec(originalPlan: SparkPlan, override val output: Seq[Attribute]) |
There was a problem hiding this comment.
Spark's EmptyRelationExec overrides generateTreeString so the eliminated logical subtree prints as a pseudo child, which is the main way to see what AQE removed. This node does not, so that subtree disappears from explain output once it converts. Here is the same query with spark.comet.exec.emptyRelation.enabled off and then on:
EmptyRelation [plan_id=1471]
+- Join LeftSemi, (_1#2 = _1#6)
:- LogicalQueryStage Project [_1#2], ShuffleQueryStage 0
...
CometEmptyRelation EmptyRelation [plan_id=1592], [_1#2]
The docstring just above already calls that subtree explanation data, so would you delegate generateTreeString to originalPlan to keep it visible?
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Correctness
Reviewed a17eabc392770abb8dec87f40893b73c28967b80 against c8ee6aef50dcb4d4f8592dec4d264f6a81a4a0c9.
Before this change, Spark 4 could replace an AQE stage with EmptyRelationExec, leaving supported parents unable to start native execution from that leaf. The new operator uses the existing scan/Arrow-input path. It retains Spark's output attributes and canonical output, returns zero partitions, and does not execute the eliminated logical subtree.
I traced this against maintained Spark 4.0 source at 03f28fc43180. Global final aggregates still receive a single-partition exchange, while shuffled joins retain aligned inputs. Broadcast inputs are expanded to the probe partition count, including reused exchanges and AQE broadcast wrappers. Existing type, aggregate-buffer, existence-join, and broadcast nested-loop restrictions remain intact. Spark's writer creates an empty task for zero-partition input, which explains why the new writer fallback is necessary.
The committed tests check both answers and native plan nodes for aggregates, 33 empty-side hash-join combinations, reuse, and AQE. The latest commit also retains all ten empty-plan assertions in each Spark test patch, including the five negative assertions. I found no new P1/P2 issue in the reviewed paths. I agree that the default-configuration reachability coverage discussed in the existing AQE comment remains useful. This change consumes empty relations but does not extend Spark's row-count inference through CometHashAggregateExec.
Validation
My checks were source/ancestry review, maintained Spark comparison, test-patch auditing, and the existing suite-registration check, which passed. I did not run the current-head Spark/JNI suites or benchmarks. The suite counts and writer-guard mutation result in the description are author reports, and the broader execution results are andygrove's review. Maintained Spark 3.4 and 4.1 sources were unavailable, so I am not claiming independent semantic coverage for those versions.
Current CI requires workflow approval and has no test jobs. The green label check is not test coverage. GitHub's cached merge also has a different first parent from the assigned base, so it is not validation of this exact pair.
Performance
The empty leaf creates no Arrow reader or per-partition work. Letting supported parents resume native execution is a plausible benefit, but no speedup was measured in this review. The writer guard walks the plan during eligibility checks and can also select Spark's writer for nonempty output with an empty descendant. That broader fallback is documented and already discussed in the writer comment. I found no additional performance regression in the reviewed code. Please include a matched microbenchmark for an empty-side join that reaches this operator at default settings, with the conversion enabled and disabled. Report the executed plans alongside timings so the intended benefit is distinguished from differences in AQE planning.
Design
Reusing the existing sink and Arrow-source interfaces keeps the implementation small and leaves distribution requirements with Spark. The writer guard preserves readable empty Parquet output without expanding this PR into the broader writer fix in #5303. This PR consumes empty relations. Extending row-count inference remains separate.
Abstraction & complexity
The Spark-version shim isolates the class introduced in Spark 4, and the input traversal now matches the same Arrow-source trait used by execution. That generalization covers the two existing source implementations and the new leaf consistently. The test-only shim factory, test placement, explain subtree, and operator-index documentation are already covered by the existing review. I have no additional requests beyond those discussions.
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Rechecked a17eabc3 → 4502095a against base c8ee6aef. The explain change preserves the eliminated subtree through innerChildren while keeping the operator a leaf; the new regression checks both the displayed subtree and empty execution. All eight moved test bodies and the three extracted writer helpers are unchanged apart from Spark-version assumptions and formatting. The Spark 4-only suites are registered in both CI workflows, the unused shim factory is removed, and the operator index and #5303 writer comment address the existing documentation requests. No new or remaining verified P1/P2 findings.
The default-configuration reachability suggestion and previously requested matched benchmark remain open. This update does not extend Spark’s row-count inference through CometHashAggregateExec; I have no additional inline requests.
The local suite-registration check and source-equivalence assertions passed. At 2026-09-11 17:48 UTC, CI remains in progress: 55 checks succeeded, 14 are running, and 7 were skipped. Linux Spark 4.2 scans explicitly passed the moved empty-Parquet-output test; its ScalaTest total was 259 passed, 247 canceled, 1 ignored, and 0 failed. The inspected jobs checked out merge b72f6b6c with first parent f29a2361, rather than the assigned base, so this is qualified merge-CI evidence, not validation of the exact pair. I ran no local build, Spark/JNI suite or benchmark. Maintained Spark 3.4/4.1 sources remain unavailable.
3 participants
Which issue does this PR close?
Closes #5819.
Rationale for this change
When Spark 4 AQE replaces an empty query stage with
EmptyRelationExec, Comet does not recognize the new leaf as a native input. Supported joins and aggregates above it can fall back to Spark.What changes are included in this PR?
CometEmptyRelationExecthrough the existing Arrow-input path, preserving Spark's output attributes and zero partitions without executing the eliminated subtree.spark.comet.exec.emptyRelation.enabled.How are these changes tested?
Eight regression tests cover attributes, canonicalization, configuration, zero partitions and an empty input stream, global/grouped COUNT and SUM, 33 empty-side hash-join combinations, aggregate-buffer and existence-join fallback, reused broadcasts, AQE, and empty Parquet overwrite/readback.
Local validation used the upstream Maven profiles and unchanged native dependencies on Linux amd64 / OpenJDK 17:
Spark 4.0 had two version-specific cancellations. Spark 3.4/3.5 had 11/9 version-specific cancellations, including the eight Spark 4-only tests. The Spark 4 aggregate suite retains its two existing ignored tests.
Also passed: native build, 129 unchanged Spark 4.0 TPC-DS plan snapshots, Maven packaging, Scalafix, Scalastyle, Spotless and Markdown formatting. Removing only the writer guard makes the new overwrite/readback test fail with
PATH_NOT_FOUND; restoring it passes.The three Spark 4.1 AQE regressions were reproduced locally at the original
EmptyRelationExecclass assertions before fixing them. Both Spark test patches were regenerated from modified Spark source and verified to apply cleanly to their release tags.